package com.example.mq.server;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.common.Request;
import com.example.mq.common.Response;
import com.example.mq.common.payloads.*;
import com.example.mq.server.virtualhost.VirtualHost;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BrokerServer {
    private ServerSocket serverSocket = null;

    private VirtualHost virtualHost = new VirtualHost("default");

    // 引入一个哈希表，存储已连接的客户端
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<String, Socket>();

    // 引入一个线程池，实现同时处理多个客户端的连接和请求
    private ExecutorService executorService = null;

    // 引入一个布尔变量，控制服务器是否继续运行
    private volatile boolean runnable = true;


    // 分配端口号
    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }


    public void start() throws IOException {
        System.out.println("服务器启动，等待客户端连接...");

        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                // 阻塞等待连接
                Socket clientSocket = serverSocket.accept();

                executorService.submit(() -> {
                    // 处理连接
                    processConnection(clientSocket);
                });

            }
        } catch (SocketException e) {
            System.out.println("服务器停止运行");
            e.printStackTrace();
        }
    }

    public void stop() throws IOException {
        runnable = false;

        // 放弃线程池中的所有任务，销毁所有线程
        executorService.shutdownNow();

        serverSocket.close();
    }

    // 处理连接和请求
    // 多个连接和请求同时处理，故接下来的业务运行需要考虑线程安全
    private void processConnection(Socket clientSocket) {
        try {
            InputStream inputStream = clientSocket.getInputStream();
            OutputStream outputStream = clientSocket.getOutputStream();
            DataInputStream dataInputStream = new DataInputStream(inputStream);
            DataOutputStream dataOutputStream = new DataOutputStream(outputStream);

            while (true) {
                // 接收并处理请求
                Request request = readRequest(dataInputStream);
                // 根据请求计算响应
                Response response = process(request, clientSocket);
                // 发送响应
                writeResponse(dataOutputStream, response);
            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (MqException e) {
            e.printStackTrace();
        }
    }

    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());

        // 刷新缓冲区
        dataOutputStream.flush();
    }

    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {



        // 1. 服务器显示收到请求的基本信息
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());

        // 这句代码容易挂
        sessions.put(basicArguments.getClientTag(), clientSocket);
        System.out.println(sessions.toString());
        virtualHost.setSessions(sessions);

        System.out.println("[BrokerServer]: 收到请求：" + "rid = " + basicArguments.getRid() + ", type=" + request.getType() + ", length=" + request.getLength());

        // 2. 根据 type 值进行分支
        boolean ok = true;
        if (request.getType() == 0x1) {
            // 创建交换机. 此时 payload 就是 ExchangeDeclareArguments 对象了.
            PayloadExchangeDeclare payloadExchangeDeclare = (PayloadExchangeDeclare) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.exchangeDeclare(payloadExchangeDeclare.getExchangeName(), payloadExchangeDeclare.getExchangeType());

        } else if (request.getType() == 0x2) {
            PayloadExchangeDelete payloadExchangeDelete = (PayloadExchangeDelete) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.exchangeDelete(payloadExchangeDelete.getExchangeName());

        } else if (request.getType() == 0x3) {
            PayloadQueueDeclare payloadQueueDeclare = (PayloadQueueDeclare) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.queueDeclare(payloadQueueDeclare.getQueueName());

        } else if (request.getType() == 0x4) {
            PayloadQueueDelete payloadQueueDelete = (PayloadQueueDelete) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.queueDelete((payloadQueueDelete.getQueueName()));

        } else if (request.getType() == 0x5) {
            PayloadBind payloadBind = (PayloadBind) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.bind(payloadBind.getQueueName(), payloadBind.getExchangeName(), payloadBind.getBindingKey());

        } else if (request.getType() == 0x6) {
            PayloadUnbind payloadUnbind = (PayloadUnbind) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.unbind(payloadUnbind.getQueueName(), payloadUnbind.getExchangeName());

        } else if (request.getType() == 0x7) {
            PayloadProduce payloadProduce = (PayloadProduce) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.produceMessage(payloadProduce.getExchangeName(), payloadProduce.getRoutingKey(), payloadProduce.getBody());

        } else if (request.getType() == 0x8) {
            PayloadSubscribe payloadSubscribe = (PayloadSubscribe) BinaryTool.fromBytes(request.getPayload());
            ok = virtualHost.processSubscribe(payloadSubscribe.getClientTag(), payloadSubscribe.getQueueName());
        } else {
            // 当前的 type 是非法的.
            throw new MqException("[BrokerServer] 未知的 type：type=" + request.getType());
        }

        // 3. 构造响应
        PayloadResponse payloadResponse = new PayloadResponse();
        payloadResponse.setRid(basicArguments.getRid());
        payloadResponse.setOk(ok);
        byte[] payload = BinaryTool.toBytes(payloadResponse);
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[BrokerServer] 返回响应：rid=" + payloadResponse.getRid() + ", type=" + response.getType() + ", length=" + response.getLength() + ", ok = " + payloadResponse.isOk());
        return response;
    }

    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();

        // readInt(), 从基础输入流中读取4个字节并将它们解释为一个int值
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());

        byte[] payload = new byte[request.getLength()];

        // 从基础输入流中读取数据，填充payload
        // 返回读取了的字节的个数
        // 此处 payload 内容类型有多种可能
        int n = dataInputStream.read(payload);
        // 格式检查
        if (n != request.getLength()) {
            throw new IOException("读取请求格式出错!");
        }

        request.setPayload(payload);
        return request;
    }
}
